package com.yahoo.ycsb.db;

import java.io.IOException;
import java.util.HashMap;
import java.util.Properties;
import java.util.Random;
import java.util.Set;
import java.util.Vector;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.basho.riak.pbc.RequestMeta;
import com.basho.riak.pbc.RiakObject;
import com.yahoo.ycsb.DB;
import com.yahoo.ycsb.DBException;
/**
 *  Riak client for YCSB framework.
 *  
 *  Properties to set:
 *  
 *  riak.hostnames = Riak server hostnames (divided by comma)
 *  //depricated: riak.bucket = Riak buckets name 
 * 
 * @author barintom
 *
 */
public class RiakClient extends DB {
	private static final Logger logger = LoggerFactory.getLogger(RiakClient.class);
	
	private com.basho.riak.pbc.RiakClient riak;
	private String host;
	private String bucket;
	private static Random random = new Random();
	

	/* (non-Javadoc)
	 * @see com.yahoo.ycsb.DB#init()
	 */
	@Override
	public void init() throws DBException {
		Properties props = getProperties();
        String hosts = props.getProperty("riak.hostname");
        String tmpHosts[] = hosts.split(",");
        host = tmpHosts[random.nextInt(tmpHosts.length)];        
        bucket = props.getProperty("riak.bucket");
        try {
			riak = new com.basho.riak.pbc.RiakClient(host);
		} catch (IOException e) {
			logger.error("Could not initialize riak",e);
			throw new DBException(e);
		}
	}

	@Override
	public int read(String table, String key, Set<String> fields,
			HashMap<String, String> result) {
		// TODO Auto-generated method stub
		return 0;
	}

	@Override
	public int scan(String table, String startkey, int recordcount,
			Set<String> fields, Vector<HashMap<String, String>> result) {
		logger.warn("Used not implemented method: scan");
		return -1;
	}

	@Override
	public int update(String table, String key, HashMap<String, String> values) {
		return insert(table,key,values);
	}

	@Override
	/**
     * Insert a record in the database. Any field/value pairs in the specified values HashMap will be written into the record with the specified
     * record key.
     *
     * @param table The name of the table
     * @param key The record key of the record to insert.
     * @param values A HashMap of field/value pairs to insert in the record
     * @return Zero on success, a non-zero error code on error. See this class's description for a discussion of error codes.
     */
	public int insert(String table, String key, HashMap<String, String> values) {
		//com.mongodb.DB db = null;
		RiakObject obj = null;
        try {
        	
        	StringBuilder sb = new StringBuilder(50000);
        /*	sb.append("{\n");
        	for (String field : values.keySet())
            {
              sb.append("\""+field+"\": "+"\""+values.get(field)+"\",\n");
            }
        	sb.append("\"\":\"\"\n}");
        	*/
        	
        	//long ms = System.currentTimeMillis();
        	obj = new RiakObject(table, key,values.toString());
        	riak.store(obj);
        	
        	//System.out.println(System.currentTimeMillis()-ms);
        	
            return 0;
        } catch (Exception e) {
            logger.error(e + "", e);
            return 1;
        } finally {
            ;
        }
	}

	@Override
	public int delete(String table, String key) {

		try {
			riak.delete(table, key);
		} catch (IOException e) {
			logger.error(e + "", e);
			return 0;
		}
		return 0;
	}

}
